package com.chb.catRecommend;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


/**
 * 第二个mapreduce 
 */
public class Step2 {
	public static boolean run(Configuration conf, Map<String, String> paths) throws Exception {
		FileSystem fs = FileSystem.get(conf);
		Job job = Job.getInstance();
		job.setJar("C:\\Users\\12285\\Desktop\\cr.jar");
		job.setJarByClass(Step2.class);
		job.setJobName("Step2");

		job.setMapperClass(Step2Mapper.class);
		job.setReducerClass(Step2Reducer.class);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);

		Path in = new Path(paths.get("step2Input"));
		FileInputFormat.addInputPath(job, in);
		Path out = new Path(paths.get("step2Output"));
		if (fs.exists(out)) {
			fs.delete(out, true);
		}
		FileOutputFormat.setOutputPath(job, out);
		boolean f = job.waitForCompletion(true);
		return f;
	}
	static class Step2Mapper extends Mapper<LongWritable, Text, Text, Text> {
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			//输入数据：item_id,user_id,action,vtime
			String item_id = value.toString().split(",")[0];
			String user_id = value.toString().split(",")[1];
			String action = value.toString().split(",")[2];

			Integer rv = RunJob.R.get(action);
			context.write(new Text(user_id), new Text(item_id+":"+rv.intValue()));
			//user_id     item_id1:评分 
		}
	}
	static class Step2Reducer extends Reducer<Text, Text, Text, Text> {
		//user_id     item_id1:评分 
		@Override
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			//一个用户分一个组，首先对同意物品的喜爱得分进行合并
			Map<String, Integer>  maps = new HashMap<String, Integer>();
			for (Text text : values) {
				String[] vs = text.toString().split(":");
				String item_id = vs[0];
				int action = Integer.parseInt(vs[1]);
				if (maps.get(item_id) == null) {
					maps.put(item_id, action);
				}else {
					int av = maps.get(item_id);
					av += action;
					maps.put(item_id, av);
				}
			}
			StringBuffer sb = new StringBuffer();
			//输出每个用户对每个产品的喜爱得分 输出到value, 以item_id:喜爱得分 追加， 以逗号分隔不同物品的喜爱得分情况
		    Set<String> keySet = maps.keySet();
		    for (Iterator<String> iterator = keySet.iterator(); iterator.hasNext();) {
				String item_id =  iterator.next();
				int action = maps.get(item_id);
				sb.append(item_id+":"+action+",");
			}
		    context.write(key, new Text(sb.toString()));
			
		}
	}

}
